package org.huel.dataproduce;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.IOException;
import java.util.Properties;
import java.util.Random;

public class kfkProducer {

    public static void main(String[] args) throws IOException {
        //1.创建配置对象，存储kafka的配置信息
        Properties properties = new Properties();
        //2.1 指定kafka生产者链接集群的ip地址及端口号
        // 服务器ip
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "43.143.125.94:9092");
        //2.2 指定key和value的序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        //3.创建生产者对象
        // 每隔2秒推送 5条数据
        while (true) {
            //3. 创建 kafka 生产者对象
            KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
            System.out.println("=====开始发送数据=====");
            try {
                //4. 调用 send 方法,发送消息
                for (int i = 0; i < 5; i++) {
                    String data = joint();
                    System.out.println("发送数据为:"+ data);
                    try {
                        kafkaProducer.send(new ProducerRecord<String, String>("attendance", data));
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
                Thread.sleep(2000); //间隔两秒再发送一组数据
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }


    }

    //拼接信息
    private static String joint() {
        Random rd = new Random();
        int randon = rd.nextInt(2);
        int randon2 = rd.nextInt(6);
        String[] stateArray = new String[]{"L", "A"}; // L: 缺勤, A: 出勤
        String state = stateArray[randon];
        String[] courseArray = new String[]{"Hadoop", "Spark", "Flink", "Hive", "HBase", "Kafka"};
        String course = courseArray[randon2];

        // 随机选择学生
        String[] part = getRandomStudent();
        if (part != null) {
            String res = String.join("\t", part[0], part[1], course, part[2], state);
            return res;
        }
        return null;
    }

    // 随机选择学生
    private static String[] getRandomStudent() {
        Random random = new Random();
        int randomIndex = random.nextInt(students.length);
        return students[randomIndex];
    }

    private static final String[][] students = {
            {"220713", "James", "202234071301"},
            {"220713", "Mary", "202234071302"},
            {"220713", "John", "202234071303"},
            {"220713", "Patricia", "202234071304"},
            {"220713", "Robert", "202234071305"},
            {"220713", "Jennifer", "202234071306"},
            {"220713", "Michael", "202234071307"},
            {"220713", "Linda", "202234071308"},
            {"220713", "William", "202234071309"},
            {"220713", "Elizabeth", "202234071310"},
            {"220713", "David", "202234071311"},
            {"220713", "Barbara", "202234071312"},
            {"220713", "Richard", "202234071313"},
            {"220713", "Susan", "202234071314"},
            {"220713", "Joseph", "202234071315"},
            {"220713", "Jessica", "202234071316"},
            {"220713", "Thomas", "202234071317"},
            {"220713", "Sarah", "202234071318"},
            {"220713", "Charles", "202234071319"},
            {"220713", "Karen", "202234071320"},
            {"220714", "Christopher", "202234071401"},
            {"220714", "Nancy", "202234071402"},
            {"220714", "Daniel", "202234071403"},
            {"220714", "Lisa", "202234071404"},
            {"220714", "Matthew", "202234071405"},
            {"220714", "Margaret", "202234071406"},
            {"220714", "Anthony", "202234071407"},
            {"220714", "Betty", "202234071408"},
            {"220714", "Mark", "202234071409"},
            {"220714", "Sandra", "202234071410"},
            {"220714", "Paul", "202234071411"},
            {"220714", "Ashley", "202234071412"},
            {"220714", "Steven", "202234071413"},
            {"220714", "Dorothy", "202234071414"},
            {"220714", "Andrew", "202234071415"},
            {"220714", "Kimberly", "202234071416"},
            {"220714", "Kenneth", "202234071417"},
            {"220714", "Emily", "202234071418"},
            {"220714", "Joshua", "202234071419"},
            {"220714", "Donna", "202234071420"},
            {"220715", "Kevin", "202234071501"},
            {"220715", "Michelle", "202234071502"},
            {"220715", "Brian", "202234071503"},
            {"220715", "Carol", "202234071504"},
            {"220715", "George", "202234071505"},
            {"220715", "Amanda", "202234071506"},
            {"220715", "Edward", "202234071507"},
            {"220715", "Melissa", "202234071508"},
            {"220715", "Ronald", "202234071509"},
            {"220715", "Deborah", "202234071510"},
            {"220715", "Timothy", "202234071511"},
            {"220715", "Stephanie", "202234071512"},
            {"220715", "Jason", "202234071513"},
            {"220715", "Rebecca", "202234071514"},
            {"220715", "Jeffrey", "202234071515"},
            {"220715", "Laura", "202234071516"},
            {"220715", "Ryan", "202234071517"},
            {"220715", "Helen", "202234071518"},
            {"220715", "Jacob", "202234071519"},
            {"220715", "Sharon", "202234071520"}
    };
}